#include "bplus_tree.h"
#include "../interface.h"
#include <cmath> // For std::abs

//#include "logger/logger.h"
BPlusTree::BPlusTree(const std::string &table_name, const std::string &index_name, CatalogManager &catalog_manager)
    : table_name_(table_name), index_name_(index_name), logger_(Logger::GetInstance())
{
    IndexSchema index_schema = catalog_manager.getIndexSchema(table_name, index_name);
    TableSchema table_schema = catalog_manager.getTableSchema(table_name);

    root_page_id_ = index_schema.root_page_id;
    is_primary_ = (index_name == table_schema.primary_key_name);
}

void BPlusTree::updateRootPageId(uint32_t new_root_id, CatalogManager &catalog_manager)
{
    root_page_id_ = new_root_id;
    catalog_manager.updateIndexRootPageId(table_name_, index_name_, new_root_id);
}

void BPlusTree::Insert(const BPlusTreeKey &key, const Row &row, CatalogManager &catalog_manager_)
{
    if (!key.has_value())
    {
        throw std::invalid_argument("The primary key cannot be NULL.");
    }
    if (!is_primary_)
    {
        throw std::logic_error("Attempted to insert a full row into a secondary index.");
    }
    std::variant<Row, BPlusTreeKey> value = row;

    // 如果根节点为空，说明树是空的，需要创建根节点
    if (root_page_id_ == INVALID_PAGE_ID)
    {
        StorageResult r = StorageSystem::allocate_page();
        if (r.status != "success")
            throw std::runtime_error("Create root page failed");
        root_page_id_ = r.page_id;
        DBPage root_page;
        root_page.initialize(root_page_id_, PageType::BPLUS_LEAF_PAGE, 0, INVALID_PAGE_ID);
        bool is_space_enough;
        root_page.insertLeaf(key, row, is_space_enough);
        root_page.setNextPageId(root_page_id_);
        root_page.setPrevPageId(root_page_id_);
        StorageSystem::write_page(root_page_id_, root_page.serialize());
        updateRootPageId(root_page_id_, catalog_manager_);
        return;
    }
    // 1. 获取根页面
    StorageResult r = StorageSystem::get_page(root_page_id_);
    if (r.status != "success")
        throw std::runtime_error("无法获取根节点");
    DBPage root_page(std::move(r.page_data));

    // 2. 从根节点开始递归插入
    auto result = InsertInternal(&root_page, key, value);

    // 3. 处理根节点分裂的特殊情况
    if (result.has_value())
    {
        // 根节点分裂了，需要创建新的根
        StorageResult r_new_root = StorageSystem::allocate_page();
        if (r_new_root.status != "success")
            throw std::runtime_error("Create new root page failed");

        DBPage new_root;
        // 新根的层级是旧根层级+1
        new_root.initialize(r_new_root.page_id, PageType::BPLUS_INTERNAL_PAGE, root_page.getLevel() + 1, INVALID_PAGE_ID);

        // 新根的两个孩子是旧根和分裂出的新页面
        // 注意：这里需要一个设置“最左指针”的方法
        new_root.setLeftmostPointer(root_page.getPageId());
        bool is_space_enough;
        new_root.insertInternal(result->first, result->second, is_space_enough);

        // 更新旧根和新兄弟的父指针
        root_page.setParentPageId(new_root.getPageId());

        StorageResult r_sibling = StorageSystem::get_page(result->second);
        DBPage sibling_page(std::move(r_sibling.page_data));
        sibling_page.setParentPageId(new_root.getPageId());

        // 写回所有修改的页面
        StorageSystem::write_page(root_page.getPageId(), root_page.serialize());
        StorageSystem::write_page(sibling_page.getPageId(), sibling_page.serialize());
        StorageSystem::write_page(new_root.getPageId(), new_root.serialize());

        // 更新树的根ID
        root_page_id_ = new_root.getPageId();

        // 更新CatalogManager中的根页面ID
        catalog_manager_.updateTablePrimaryPageId(table_name_, new_root.getPageId());
    }
    else
    {
        // 如果根节点没有分裂，但可能被修改了，写回
        StorageSystem::write_page(root_page.getPageId(), root_page.serialize());
    }
}

void BPlusTree::Insert(const BPlusTreeKey &key, const BPlusTreeKey &value, CatalogManager &catalog_manager)
{
    if (!key.has_value())
    {
        throw std::invalid_argument("The secondary key cannot be NULL.");
    }
    if (is_primary_)
    {
        throw std::logic_error("Attempted to insert a key-value pair into a primary index.");
    }
    std::variant<Row, BPlusTreeKey> variant_value = value;

    if (root_page_id_ == INVALID_PAGE_ID)
    {
        StorageResult r = StorageSystem::allocate_page();
        if (r.status != "success")
            throw std::runtime_error("Create root page failed");
        uint32_t new_root_id = r.page_id;
        DBPage root_page;
        root_page.initialize(new_root_id, PageType::BPLUS_LEAF_PAGE, 0, INVALID_PAGE_ID);
        bool is_space_enough;
        root_page.insertLeaf(key, value, is_space_enough);
        root_page.setNextPageId(new_root_id);
        root_page.setPrevPageId(new_root_id);
        StorageSystem::write_page(new_root_id, root_page.serialize());
        updateRootPageId(new_root_id, catalog_manager);
        return;
    }

    auto root_page = getPage(root_page_id_);
    auto result = InsertInternal(root_page.get(), key, variant_value);

    if (result.has_value())
    {
        StorageResult r_new_root = StorageSystem::allocate_page();
        if (r_new_root.status != "success")
            throw std::runtime_error("Create new root page failed");
        DBPage new_root;
        new_root.initialize(r_new_root.page_id, PageType::BPLUS_INTERNAL_PAGE, root_page->getLevel() + 1, INVALID_PAGE_ID);
        new_root.setLeftmostPointer(root_page->getPageId());
        bool is_space_enough;
        new_root.insertInternal(result->first, result->second, is_space_enough);
        root_page->setParentPageId(new_root.getPageId());
        auto sibling_page = getPage(result->second);
        sibling_page->setParentPageId(new_root.getPageId());
        StorageSystem::write_page(root_page->getPageId(), root_page->serialize());
        StorageSystem::write_page(sibling_page->getPageId(), sibling_page->serialize());
        StorageSystem::write_page(new_root.getPageId(), new_root.serialize());
        updateRootPageId(new_root.getPageId(), catalog_manager);
    }
    else
    {
        StorageSystem::write_page(root_page->getPageId(), root_page->serialize());
    }
}

std::optional<std::pair<BPlusTreeKey, uint32_t>> BPlusTree::InsertInternal(DBPage *page, const BPlusTreeKey &key, const std::variant<Row, BPlusTreeKey> &value)
{
    // Print();
    if (page->isLeaf())
    {
        // --- 到达叶子节点 ---

        // 步骤 1: 预先检查空间是否足够
        bool is_space_enough = page->checkSpaceForLeafInsertion(key, value);

        if (is_space_enough)
        {
            // 步骤 2: 空间足够，直接插入
            std::visit([&](auto &&arg)
                       { page->insertLeaf(key, arg, is_space_enough); }, value);
            // 插入成功，无需向上传递，直接返回
            return std::nullopt;
        }

        // 步骤 3: 空间不足，需要分裂
        // ===================================================================
        // 核心修正逻辑：先分裂，再插入
        // ===================================================================

        // 3a. 先分裂当前已满的页面 (注意：新记录<key, row>还未插入)
        auto [promoted_key, new_page_id] = SplitPage(page);

        // 3b. 决定新记录应该插入到哪个页面
        // promoted_key 是新页面的第一个键。所有小于它的键都留在旧页。
        if (compareKeys(key, promoted_key) < 0)
        {
            // 插入到旧页面
            bool is_space_enough;
            std::visit([&](auto &&arg)
                       { page->insertLeaf(key, arg, is_space_enough); }, value);
            // 分裂后，旧页面肯定有空间，所以 is_space_enough 必然为 true
        }
        else
        {
            // 插入到新页面
            auto new_page = getPage(new_page_id);
            if (!new_page)
                throw std::runtime_error("can't get new page after split");

            bool is_space_enough;
            std::visit([&](auto &&arg)
                       { new_page->insertLeaf(key, arg, is_space_enough); }, value);
            StorageSystem::write_page(new_page->getPageId(), new_page->serialize());
        }

        // 3c. 分裂已经发生，返回需要提拔的键和新页面ID给上层
        return {{promoted_key, new_page_id}};
    }
    else
    {
        // --- 内部节点 ---
        // 1. 找到应该进入的子节点
        uint32_t child_page_id = page->findChildPage(key);

        auto child_page = getPage(child_page_id);
        if (!child_page)
            throw std::runtime_error("can't get child page");

        // 2. 递归调用插入
        auto result = InsertInternal(child_page.get(), key, value);

        // 3. 处理子节点的分裂
        if (result.has_value())
        {
            // 子节点分裂了，需要将提拔的键插入到当前节点
            auto [promoted_key, new_child_page_id] = result.value();

            // 预先检查当前内部节点的空间
            bool is_space_enough = page->checkSpaceForInternalInsertion(promoted_key);

            if (is_space_enough)
            {
                // 空间足够，直接插入
                page->insertInternal(promoted_key, new_child_page_id, is_space_enough);
                StorageSystem::write_page(child_page->getPageId(), child_page->serialize()); // 写回被修改的子节点
                return std::nullopt;                                                         // 任务完成，无需再向上传递
            }

            // 空间不足，当前内部节点也需要分裂
            // ===================================================================
            // 核心修正逻辑：先分裂，再插入
            // ===================================================================

            // 1. 分裂当前内部节点
            auto [key_to_grandparent, new_sibling_page_id] = SplitPage(page);

            // 2. 决定 <promoted_key, new_child_page_id> 应该插入到哪个分裂后的节点
            if (compareKeys(promoted_key, key_to_grandparent) < 0) // 使用 compareKeys
            {
                // 插入到旧节点
                bool is_space_enough;
                page->insertInternal(promoted_key, new_child_page_id, is_space_enough);
            }
            else
            {
                // 插入到新兄弟节点
                auto new_sibling_page = getPage(new_sibling_page_id);
                if (!new_sibling_page)
                {
                    throw std::runtime_error("can't get new sibling page after split");
                }
                bool is_space_enough;
                new_sibling_page->insertInternal(promoted_key, new_child_page_id, is_space_enough);
                StorageSystem::write_page(new_sibling_page->getPageId(), new_sibling_page->serialize());
            }

            // 写回被修改的子节点
            StorageSystem::write_page(child_page->getPageId(), child_page->serialize());

            // 3. 将从当前层提拔的键 (key_to_grandparent) 返回给上层（祖父节点）
            return {{key_to_grandparent, new_sibling_page_id}};
        }
        else
        {
            // 子节点没有分裂
            StorageSystem::write_page(child_page->getPageId(), child_page->serialize());
            return std::nullopt;
        }
    }
}

std::pair<BPlusTreeKey, uint32_t> BPlusTree::SplitPage(DBPage *page_to_split)
{
    // 1. 分配新页
    StorageResult r_new = StorageSystem::allocate_page();
    if (r_new.status != "success")
        throw std::runtime_error("Allocate new page failed");

    DBPage new_page;
    new_page.initialize(r_new.page_id, page_to_split->getPageType(), page_to_split->getLevel(), page_to_split->getParentPageId());

    // 2. 移动一半数据
    auto promoted_key_opt = page_to_split->MoveHalfTo(&new_page);

    BPlusTreeKey promoted_key;
    if (page_to_split->isLeaf())
    {
        // 叶子分裂，复制新页的第一个键
        promoted_key = new_page.getKey(0);

        // 更新叶子链表
        new_page.setNextPageId(page_to_split->getNextPageId());
        if (page_to_split->getNextPageId() != INVALID_PAGE_ID)
        {
            auto r_next = StorageSystem::get_page(page_to_split->getNextPageId());
            DBPage next_page(std::move(r_next.page_data));
            next_page.setPrevPageId(new_page.getPageId());
            StorageSystem::write_page(next_page.getPageId(), next_page.serialize());
        }
        page_to_split->setNextPageId(new_page.getPageId());
        new_page.setPrevPageId(page_to_split->getPageId());
    }
    else
    {
        // 内部分裂，使用 MoveHalfTo 返回的被提拔的键
        promoted_key = promoted_key_opt.value();

        // 更新被移动到新页的子节点的父指针
        for (int i = 0; i < new_page.getNumRecords(); ++i)
        {
            uint32_t child_id = new_page.getInternalValue(i);
            auto r_child = StorageSystem::get_page(child_id);
            DBPage child_page(std::move(r_child.page_data));
            child_page.setParentPageId(new_page.getPageId());
            StorageSystem::write_page(child_id, child_page.serialize());
        }
        // 更新最左边的子节点的父指针
        uint32_t leftmost_child_id = new_page.getLeftmostPointer();
        auto r_leftmost_child = StorageSystem::get_page(leftmost_child_id);
        DBPage leftmost_child_page(std::move(r_leftmost_child.page_data));
        leftmost_child_page.setParentPageId(new_page.getPageId());
        StorageSystem::write_page(leftmost_child_id, leftmost_child_page.serialize());
    }

    // 3. 写回新页
    StorageSystem::write_page(new_page.getPageId(), new_page.serialize());

    // 4. 返回提拔的键和新页ID
    return {promoted_key, new_page.getPageId()};
}

uint32_t BPlusTree::Delete(const BPlusTreeKey &key, CatalogManager &catalog_manager_)
{
    if (!key.has_value())
    {
        // 删除一个 NULL 键不会影响任何行
        return 0;
    }
    if (!is_primary_)
    {
        throw std::logic_error("Called primary Delete on a secondary index. Use Delete(key, value).");
    }
    // 增加了一个模板参数，用于区分叶子节点和内部节点的删除
    return DeleteInternal(nullptr, root_page_id_, key, nullptr, catalog_manager_);
}
uint32_t BPlusTree::Delete(const BPlusTreeKey &key, const BPlusTreeKey &value, CatalogManager &catalog_manager)
{
    if (!key.has_value())
    {
        // 删除一个 NULL 键不会影响任何行
        return 0;
    }
    if (is_primary_)
    {
        throw std::logic_error("Called secondary Delete on a primary index. Use Delete(key).");
    }
    return DeleteInternal(nullptr, root_page_id_, key, &value, catalog_manager);
}
// 递归删除函数
// parent_page: 当前节点的父节点
// current_page_id: 当前节点的ID
// key: 要删除的键
// 返回值：是否成功删除了key
uint32_t BPlusTree::DeleteInternal(DBPage *parent_page, uint32_t current_page_id, const BPlusTreeKey &key, const BPlusTreeKey *value_ptr, CatalogManager &catalog_manager)
{
    auto current_page = getPage(current_page_id);
    bool key_deleted = false;
    uint32_t delete_count = 0;
    if (current_page->isLeaf())
    {
        int index = current_page->findLeafKeyIndex(key);
        if (index != -1)
        {
            if (is_primary_)
            {
                current_page->DeleteRecord(index);
                delete_count++;
                key_deleted = true;
                StorageSystem::write_page(current_page->getPageId(), current_page->serialize());
            }
            else
            {
                if (value_ptr == nullptr)
                {
                    throw std::invalid_argument("Secondary index delete requires a value to be specified.");
                }
                int num_records = current_page->getNumRecords();
                for (int i = index; i < num_records && current_page->getKey(i) == key; ++i)
                {
                    if (current_page->getLeafValue(i) == *value_ptr)
                    {
                        current_page->DeleteRecord(i);
                        key_deleted = true;
                        delete_count++;
                        break;
                    }
                }
            }
        }
    }
    else
    { // 内部节点
        // 找到应该去往的子节点，并递归删除
        uint32_t child_page_id = current_page->findChildPage(key);
        key_deleted = DeleteInternal(current_page.get(), child_page_id, key, value_ptr, catalog_manager);
    }

    if (!key_deleted)
    {
        return delete_count; // 键未找到，无需任何操作
    }

    // --- 核心逻辑：删除后检查下溢 ---

    // 如果是根节点，特殊处理
    if (current_page->isRoot())
    {
        // 如果根节点在删除后变为空（通常是内部根节点的孩子被合并）
        if (!current_page->isLeaf() && current_page->getNumRecords() == 0)
        {
            // 树的高度需要降低
            root_page_id_ = current_page->getLeftmostPointer();
            // 更新新根的父指针
            auto new_root = getPage(root_page_id_);
            new_root->setParentPageId(INVALID_PAGE_ID);
            StorageSystem::write_page(new_root->getPageId(), new_root->serialize());
            // 更新CatalogManager中的根页面ID
            updateRootPageId(root_page_id_, catalog_manager);
            // 删除旧的根页面
            StorageSystem::free_page(current_page->getPageId());
        }
        else if (current_page->isLeaf() && current_page->getNumRecords() == 0)
        {
            // 整棵树变空了
            StorageSystem::free_page(current_page->getPageId());
            root_page_id_ = INVALID_PAGE_ID;
            catalog_manager.updateTablePrimaryPageId(table_name_, root_page_id_);
        }
        return delete_count;
    }

    // 非根节点，检查是否下溢
    if (current_page->IsUnderflow())
    {
        HandleUnderflow(parent_page, current_page.get());
    }
    else
    {
        // 即使不下溢，页面也被修改了，需要写回
        StorageSystem::write_page(current_page->getPageId(), current_page->serialize());
    }
    return delete_count;
}
void BPlusTree::HandleUnderflow(DBPage *parent_page, DBPage *underflow_page)
{
    // 1. 找到当前节点在父节点中的索引，以及它的左右兄弟
    int child_index_in_parent = parent_page->findChildIndex(underflow_page->getPageId());
    if (child_index_in_parent == -2)
    {
        throw std::runtime_error("Cannot find brother pages on the parent page.");
    }
    uint32_t left_sibling_id = INVALID_PAGE_ID;
    uint32_t right_sibling_id = INVALID_PAGE_ID;

    if (child_index_in_parent == -1)
    { // (P_leftmost)
        // 没有左兄弟
        if (parent_page->getNumRecords() > 0)
        {
            right_sibling_id = parent_page->getInternalValue(0);
        }
    }
    else
    { // (P_i, i >= 0)
        if (child_index_in_parent == 0)
        {
            left_sibling_id = parent_page->getLeftmostPointer();
        }
        else
        {
            left_sibling_id = parent_page->getInternalValue(child_index_in_parent - 1);
        }

        if (child_index_in_parent < parent_page->getNumRecords() - 1)
        {
            right_sibling_id = parent_page->getInternalValue(child_index_in_parent + 1);
        }
    }

    // 2. 优先尝试从左兄弟重分配
    if (left_sibling_id != INVALID_PAGE_ID)
    {
        auto left_sibling = getPage(left_sibling_id);
        // 检查左兄弟是否“富余”，即记录数大于最小阈值
        if (left_sibling->isSurplus())
        {
            Redistribute(left_sibling.get(), underflow_page, parent_page, child_index_in_parent);
            return; // 重分配完成
        }
    }

    // 3. 尝试从右兄弟重分配
    if (right_sibling_id != INVALID_PAGE_ID)
    {
        auto right_sibling = getPage(right_sibling_id);
        if (right_sibling->isSurplus())
        {
            Redistribute(right_sibling.get(), underflow_page, parent_page, child_index_in_parent, false);
            return; // 重分配完成
        }
    }

    // 4. 重分配失败，必须合并
    // 优先与左兄弟合并
    if (left_sibling_id != INVALID_PAGE_ID)
    {
        Merge(parent_page, left_sibling_id, underflow_page->getPageId(), child_index_in_parent);
    }
    else
    {
        // 否则与右兄弟合并
        Merge(parent_page, underflow_page->getPageId(), right_sibling_id, child_index_in_parent + 1);
    }
}
void BPlusTree::Redistribute(DBPage *sibling_page, DBPage *page_in_need, DBPage *parent_page, int index_in_parent, bool is_left_sibling)
{
    if (is_left_sibling)
    {
        // 左兄弟 -> 右边（当前节点）
        // 1. 将父节点中的分隔键下移到当前节点
        BPlusTreeKey key_from_parent = parent_page->getKey(index_in_parent - 1);

        // 2. 将左兄弟的最后一个记录移动到当前节点
        if (page_in_need->isLeaf())
        {
            // 叶子节点：直接移动记录，并用兄弟的新最大键更新父节点
            sibling_page->MoveRecordFrom(page_in_need, sibling_page->getNumRecords() - 1, 0);
            parent_page->setKey(index_in_parent, page_in_need->getFirstKey());
        }
        else
        {
            // 内部节点：父键下移，兄弟的最右孩子指针成为当前节点的最左指针
            int last_index = sibling_page->getNumRecords() - 1;
            uint32_t sibling_last_pointer = sibling_page->getInternalValue(last_index);
            bool is_space_enough;
            page_in_need->insertInternal(0, key_from_parent, page_in_need->getLeftmostPointer(), is_space_enough);
            page_in_need->setLeftmostPointer(sibling_last_pointer);
            parent_page->setKey(index_in_parent, sibling_page->getLastKey());
            sibling_page->DeleteRecord(last_index);
        }
    }
    else
    {
        // 右兄弟 -> 左边（当前节点）
        // 1. 将父节点中的分隔键下移到当前节点
        BPlusTreeKey key_from_parent = parent_page->getKey(index_in_parent);

        // 2. 将右兄弟的第一个记录移动到当前节点
        if (page_in_need->isLeaf())
        {
            // 叶子节点：直接移动记录，并用兄弟的新最大键更新父节点
            sibling_page->MoveRecordFrom(page_in_need, 0, page_in_need->getNumRecords());
            parent_page->setKey(index_in_parent + 1, sibling_page->getFirstKey());
        }
        else
        {
            // 父键下移，兄弟的最左孩子指针成为当前节点的最右指针
            bool is_space_enough;
            page_in_need->insertInternal(page_in_need->getNumRecords(), key_from_parent, sibling_page->getLeftmostPointer(), is_space_enough);
            parent_page->setKey(index_in_parent + 1, sibling_page->getFirstKey());
            sibling_page->setLeftmostPointer(sibling_page->getInternalValue(0));
            sibling_page->DeleteRecord(0);
        }
    }
    StorageSystem::write_page(sibling_page->getPageId(), sibling_page->serialize());
    StorageSystem::write_page(page_in_need->getPageId(), page_in_need->serialize());
    StorageSystem::write_page(parent_page->getPageId(), parent_page->serialize());
}

void BPlusTree::Merge(DBPage *parent_page, uint32_t left_page_id, uint32_t right_page_id, int index_in_parent)
{
    auto left_page = getPage(left_page_id);
    auto right_page = getPage(right_page_id);

    if (left_page->isLeaf())
    {
        // 叶子节点合并：将右节点的所有记录移动到左节点，并更新链表
        right_page->MoveAllTo(left_page.get());
        left_page->setNextPageId(right_page->getNextPageId());
        // 更新后继节点的 prev 指针
        if (right_page->getNextPageId() != INVALID_PAGE_ID)
        {
            auto successor = getPage(right_page->getNextPageId());
            successor->setPrevPageId(left_page->getPageId());
            StorageSystem::write_page(successor->getPageId(), successor->serialize());
        }
    }
    else
    {
        // 内部节点合并：将父节点的分隔键下移到左节点，然后将右节点的所有记录和指针移动到左节点
        BPlusTreeKey key_from_parent = parent_page->getKey(index_in_parent); // 假设index是分隔键的索引
        bool is_space_enough;
        int left_num_records = left_page->getNumRecords();
        left_page->insertInternal(left_num_records, key_from_parent, right_page->getLeftmostPointer(), is_space_enough);
        right_page->MoveAllTo(left_page.get());
        updateChildrenParent(left_page.get(), left_num_records, left_page->getNumRecords());
    }

    // 从父节点中删除指向右节点的键和指针
    parent_page->DeleteRecord(index_in_parent);

    StorageSystem::write_page(parent_page->getPageId(), parent_page->serialize());
    StorageSystem::write_page(left_page->getPageId(), left_page->serialize());

    // 回收 right_page
    StorageSystem::free_page(right_page->getPageId());
}

std::unique_ptr<DBPage> BPlusTree::getPage(uint32_t page_id)
{
    StorageResult r = StorageSystem::get_page(page_id);
    if (r.status != "success" || r.page_data.empty())
    {
        throw std::runtime_error("Failed to get page: " + std::to_string(page_id));
    }
    // 使用 make_unique 构造 DBPage 对象，并将其包装在 unique_ptr 中返回
    // std::move 用于转移 page_data 的所有权，避免拷贝
    return std::make_unique<DBPage>(std::move(r.page_data));
}

void BPlusTree::updateChildrenParent(DBPage *parent_page, int start_index, int end_index)
{
    for (int i = start_index; i < end_index; ++i)
    {
        uint32_t child_id = parent_page->getInternalValue(i);
        auto child_page = getPage(child_id);
        child_page->setParentPageId(parent_page->getPageId());
        StorageSystem::write_page(child_id, child_page->serialize());
    }
}

/**
 * @brief 公共接口：打印B+树。
 */
void BPlusTree::Print()
{
    std::cout << "================ B+ Tree Structure ================" << std::endl;
    if (root_page_id_ == INVALID_PAGE_ID)
    {
        std::cout << "[EMPTY]" << std::endl;
    }
    else
    {
        PrintRecursive(root_page_id_, 0, true);
    }
    std::cout << "===================================================" << std::endl;
}

/**
 * @brief 递归打印B+树的辅助函数。
 * @param page_id 当前要打印的节点的页面ID。
 * @param level 当前节点的递归深度（用于缩进）。
 * @param is_last 当前节点是否是其父节点中的最后一个孩子（用于打印漂亮的连接线）。
 */
void BPlusTree::PrintRecursive(uint32_t page_id, int level, bool is_last)
{
    // 1. 获取页面数据
    // 注意：这里的 getPage 返回的是裸指针，需要手动 delete。
    // 在生产代码中，应使用智能指针。
    auto page = getPage(page_id);
    if (!page)
    {
        return;
    }

    // 2. 打印当前节点的缩进和前缀
    for (int i = 0; i < level; ++i)
    {
        std::cout << "    ";
    }
    // 打印连接线
    std::cout << (is_last ? "└── " : "├── ");

    // 3. 打印当前节点的信息
    if (page->isLeaf())
    {
        std::cout << "[LEAF " << page->getPageId() << "] Keys: {";
        for (int i = 0; i < page->getNumRecords(); ++i)
        {
            // *** MODIFIED PRINTING LOGIC ***
            const auto &key = page->getKey(i);
            if (!key.has_value())
            {
                std::cout << "NULL";
            }
            else
            {
                // 使用 valueToString 辅助函数 (假设已定义) 或者直接用 visit
                std::visit([](auto &&arg)
                           { std::cout << arg; }, key.value());
            }

            if (i < page->getNumRecords() - 1)
                std::cout << ", ";
        }
        std::cout << "} | Prev: " << page->getPrevPageId() << " | Next: " << page->getNextPageId() << std::endl;
    }
    else
    { // 内部节点
        std::cout << "[INTERNAL " << page->getPageId() << "] Keys: {";
        for (int i = 0; i < page->getNumRecords(); ++i)
        {
            // *** MODIFIED PRINTING LOGIC ***
            const auto &key = page->getKey(i);
            if (!key.has_value())
            {
                std::cout << "NULL";
            }
            else
            {
                std::visit([](auto &&arg)
                           { std::cout << arg; }, key.value());
            }

            if (i < page->getNumRecords() - 1)
                std::cout << ", ";
        }
        std::cout << "}" << std::endl;

        // 4. 递归打印子节点

        // 获取所有子节点的ID
        std::vector<uint32_t> child_page_ids;
        child_page_ids.push_back(page->getLeftmostPointer());
        for (int i = 0; i < page->getNumRecords(); ++i)
        {
            child_page_ids.push_back(page->getInternalValue(i));
        }

        // 递归调用
        for (size_t i = 0; i < child_page_ids.size(); ++i)
        {
            PrintRecursive(child_page_ids[i], level + 1, (i == child_page_ids.size() - 1));
        }
    }
}

/**
 * @brief 从根节点开始，查找到包含指定键的叶子页面。
 * @param key 目标键。
 * @return std::unique_ptr<DBPage> 指向包含该键的叶子页面的智能指针，如果树为空或未找到则为 nullptr。
 */
std::unique_ptr<DBPage> BPlusTree::FindLeafPage(const BPlusTreeKey &key)
{
    if (root_page_id_ == INVALID_PAGE_ID)
    {
        return nullptr; // 树是空的
    }

    uint32_t current_page_id = root_page_id_;
    auto current_page = getPage(current_page_id);

    // 循环直到我们到达叶子节点
    while (current_page && !current_page->isLeaf())
    {
        // 在内部节点中找到下一个要去的子页面ID
        uint32_t child_page_id = current_page->findChildPage(key);

        // 移动到子节点
        current_page_id = child_page_id;
        current_page = getPage(current_page_id);
    }

    // 返回找到的叶子页面（或者如果中途失败则返回 nullptr）
    return current_page;
}

/**
 * @brief 单点查询：根据键查找单条记录。
 */
bool BPlusTree::Search(const BPlusTreeKey &key, Row &row)
{
    // 1. 找到可能包含该键的叶子页面
    auto leaf_page = FindLeafPage(key);

    if (!leaf_page)
    {
        return false; // 树为空或查找路径中断
    }

    // 2. 在叶子页面内查找键的索引
    int index_in_leaf = leaf_page->findLeafKeyIndex(key);

    if (index_in_leaf == -1)
    {
        return false; // 在叶子中未找到该键
    }

    // 3. 找到键，反序列化对应的行数据
    return leaf_page->getLeafValue(index_in_leaf, row);
}

bool BPlusTree::SearchForSecondary(const BPlusTreeKey &key, std::vector<BPlusTreeKey> &values)
{
    if (is_primary_)
        throw std::logic_error("Search for Primary Keys called on a primary index.");
    values.clear();
    auto leaf_page = FindLeafPage(key);
    if (!leaf_page)
        return false;

    int index = leaf_page->findLeafKeyIndex(key);
    if (index == -1)
        return false;
    auto first_leaf_page_id = findFirstLeafPageId();
    while (leaf_page)
    {
        int num_records = leaf_page->getNumRecords();
        for (int i = index; i < num_records && leaf_page->getKey(i) == key; ++i)
        {
            values.push_back(leaf_page->getLeafValue(i));
        }
        if (leaf_page->getNextPageId() == INVALID_PAGE_ID || leaf_page->getNextPageId() == first_leaf_page_id)
        {
            leaf_page = nullptr;
        }
        else
        {
            leaf_page = getPage(leaf_page->getNextPageId());
        }
        index = 0;
    }
    return !values.empty();
}
uint32_t BPlusTree::findFirstLeafPageId()
{
    if (root_page_id_ == INVALID_PAGE_ID)
    {
        return INVALID_PAGE_ID;
    }
    auto current_page = getPage(root_page_id_);
    while (current_page && !current_page->isLeaf())
    {
        uint32_t child_id = current_page->getLeftmostPointer();
        current_page = getPage(child_id);
    }
    return current_page ? current_page->getPageId() : INVALID_PAGE_ID;
}

/**
 * @brief 范围查询：查找一个闭区间 [start_key, end_key] 内的所有记录。
 */
std::variant<std::vector<Row>, std::vector<BPlusTreeKey>> BPlusTree::SearchRange(const BPlusTreeKey &start_key, const BPlusTreeKey &end_key, bool left_inclusive, bool right_inclusive)
{
    std::variant<std::vector<Row>, std::vector<BPlusTreeKey>> results;
    std::vector<Row> rows;
    std::vector<BPlusTreeKey> keys;
    // 1. 找到范围开始的叶子页面
    auto current_page = start_key.has_value() ? FindLeafPage(start_key) : getPage(findFirstLeafPageId());
    if (!current_page)
    {
        return results; // 树为空
    }

    // 2. 在起始页面中，找到第一个大于等于 start_key 的记录的索引
    int current_index = 0;
    int num_records = current_page->getNumRecords();
    if (start_key.has_value())
    {
        while (current_index < num_records)
        {
            auto current_key = current_page->getKey(current_index);
            int cmp = compareKeys(current_key, start_key);

            if (left_inclusive && cmp >= 0) // current_key >= start_key
            {
                break;
            }
            else if (!left_inclusive && cmp > 0) // current_key > start_key
            {
                break;
            }
            current_index++;
        }
    }

    // 3. 开始遍历，从当前页面和当前索引开始
    auto first_leaf_page_id = findFirstLeafPageId();
    bool done = false;
    while (current_page && !done)
    {
        num_records = current_page->getNumRecords();
        // 遍历当前页面的所有相关记录
        for (; current_index < num_records; ++current_index)
        {
            BPlusTreeKey current_key = current_page->getKey(current_index);

            // 检查是否超出范围上限
            if (end_key.has_value())
            {
                int cmp = compareKeys(current_key, end_key);
                if (right_inclusive && cmp > 0) // current_key > end_key
                {
                    done = true;
                    break;
                }
                else if (!right_inclusive && cmp >= 0) // current_key >= end_key
                {
                    done = true;
                    break;
                }
            }

            // 键在范围内，获取行数据并添加到结果集
            if (is_primary_)
            {
                Row row;
                if (current_page->getLeafValue(current_index, row))
                {
                    rows.push_back(row);
                }
            }
            else
            {
                BPlusTreeKey value;
                value = current_page->getLeafValue(current_index);
                keys.push_back(value);
            }
        }

        if (done)
        {
            break;
        }

        // 移动到下一个兄弟页面
        uint32_t next_page_id = current_page->getNextPageId();
        if (next_page_id == INVALID_PAGE_ID || next_page_id == first_leaf_page_id)
        {
            break;
        }
        current_page = getPage(next_page_id);
        current_index = 0; // 从新页面的开头开始
    }
    if (is_primary_)
    {
        results = rows;
    }
    else
    {
        results = keys;
    }
    return results;
}

/**
 * @brief 全表扫描：获取表中的所有记录。
 */
std::variant<std::vector<Row>, std::vector<BPlusTreeKey>> BPlusTree::ScanAll()
{
    std::variant<std::vector<Row>, std::vector<BPlusTreeKey>> results;
    std::vector<Row> rows;
    std::vector<BPlusTreeKey> keys;
    // 1. 找到最左边的叶子页面
    auto first_leaf_page_id = findFirstLeafPageId();
    uint32_t current_page_id = first_leaf_page_id;
    if (current_page_id == INVALID_PAGE_ID)
    {
        return results; // 树为空
    }
    auto current_page = getPage(current_page_id);

    // 2. 顺着叶子节点的双向链表遍历
    while (current_page)
    {
        int num_records = current_page->getNumRecords();
        // 提取当前页面的所有行
        for (int i = 0; i < num_records; ++i)
        {
            if (is_primary_)
            {
                Row row;
                if (current_page->getLeafValue(i, row))
                {
                    rows.push_back(row);
                }
            }
            else
            {
                BPlusTreeKey value = current_page->getLeafValue(i);
                keys.push_back(value);
            }
        }

        // 移动到下一个页面
        uint32_t next_page_id = current_page->getNextPageId();
        // 检查是否遍历完成（到达链表末尾，或者在只有根节点的树中回到自身）
        if (next_page_id == INVALID_PAGE_ID || next_page_id == first_leaf_page_id)
        {
            break;
        }
        current_page = getPage(next_page_id);
    }
    if (is_primary_)
    {
        results = rows;
    }
    else
    {
        results = keys;
    }
    return results;
}

uint32_t BPlusTree::DeleteAll(CatalogManager &catalog_manager_)
{
    // 从根节点开始，递归删除所有页面
    uint32_t delete_count = 0;
    if (root_page_id_ == INVALID_PAGE_ID)
        return delete_count;

    auto root_page = getPage(root_page_id_);
    if (!root_page)
        return delete_count;

    // 递归删除所有子页面
    std::function<void(std::unique_ptr<DBPage>)> delete_pages = [&](std::unique_ptr<DBPage> page)
    {
        if (!page)
            return;
        int num_records = page->getNumRecords();
        if (!page->isLeaf())
        {
            for (int i = 0; i < num_records; ++i)
            {
                uint32_t child_page_id = page->getInternalValue(i);
                delete_pages(getPage(child_page_id));
            }
            delete_pages(getPage(page->getLeftmostPointer()));
        }
        if (page->isLeaf())
            delete_count += num_records;
        // 删除当前页面
        StorageSystem::free_page(page->getPageId());
    };

    delete_pages(std::move(root_page));
    root_page_id_ = INVALID_PAGE_ID;
    catalog_manager_.updateTablePrimaryPageId(table_name_, INVALID_PAGE_ID);
    return delete_count;
}

uint32_t BPlusTree::DeleteRange(const BPlusTreeKey &start_key, const BPlusTreeKey &end_key, CatalogManager &catalog_manager_)
{
    // 1. 找到起始叶子页面
    uint32_t delete_count = 0;
    auto first_leaf_page_id = findFirstLeafPageId();
    auto current_page = FindLeafPage(start_key);
    if (!current_page)
    {
        return delete_count; // 未找到页面
    }
    int current_index = 0;
    int num_records = current_page->getNumRecords();
    while (current_index < num_records && current_page->getKey(current_index) < start_key)
    {
        current_index++;
    }
    // 2. 删除范围内的所有记录
    while (current_page)
    {
        num_records = current_page->getNumRecords();
        for (int i = current_index; i < num_records; ++i)
        {
            BPlusTreeKey key = current_page->getKey(i);
            if (key > end_key)
            {
                return delete_count; // 超出范围，停止删除
            }
            std::unique_ptr<DBPage> parent_page;
            try
            {
                parent_page = getPage(current_page->getParentPageId());
            }
            catch (std::runtime_error &e)
            {
                logger_.Error("DeleteRange Error: " + std::string(e.what()));
                parent_page = nullptr;
            }
            delete_count += DeleteInternal(parent_page.get(), current_page->getPageId(), key, nullptr, catalog_manager_);
        }

        // 移动到下一个页面
        if (current_page->getNextPageId() == INVALID_PAGE_ID || current_page->getNextPageId() == first_leaf_page_id)
        {
            break;
        }
        current_page = getPage(current_page->getNextPageId());
        current_index = 0;
    }

    return delete_count;
}
uint32_t BPlusTree::Update(const BPlusTreeKey &key, const std::unordered_map<std::string, Value> &new_values, CatalogManager &catalog_manager_)
{

    if (!is_primary_)
    {
        throw std::logic_error("Update can only be directly called on a primary index.");
    }
    uint32_t update_count = 0;
    // 1. 找到叶子页面
    auto leaf_page = FindLeafPage(key);
    if (!leaf_page)
    {
        return update_count; // 未找到页面
    }

    // 2. 更新记录
    int key_index = leaf_page->findLeafKeyIndex(key);
    if (key_index == -1)
    {
        return false; // 未找到键
    }
    update_count += UpdateInternal(leaf_page.get(), key_index, new_values, catalog_manager_);
    // 3. 更新元数据
    StorageSystem::write_page(leaf_page->getPageId(), leaf_page->serialize());
    return update_count;
}
uint32_t BPlusTree::UpdateRange(const BPlusTreeKey &start_key, const BPlusTreeKey &end_key, const std::unordered_map<std::string, Value> &new_values, CatalogManager &catalog_manager_)
{
    // 1. 找到起始叶子页面
    uint32_t update_count = 0;
    auto first_leaf_page_id = findFirstLeafPageId();
    auto current_page = FindLeafPage(start_key);
    if (!current_page)
    {
        return update_count; // 未找到页面
    }
    int current_index = 0;
    int num_records = current_page->getNumRecords();
    while (current_index < num_records && current_page->getKey(current_index) < start_key)
    {
        current_index++;
    }
    // 2. 更新范围内的所有记录
    while (current_page)
    {
        num_records = current_page->getNumRecords();
        for (int i = current_index; i < num_records; ++i)
        {
            BPlusTreeKey key = current_page->getKey(i);
            if (key > end_key)
            {
                return true; // 超出范围，停止更新
            }
            update_count += UpdateInternal(current_page.get(), i, new_values, catalog_manager_);
        }

        // 移动到下一个页面
        if (current_page->getNextPageId() == INVALID_PAGE_ID || current_page->getNextPageId() == first_leaf_page_id)
        {
            break;
        }
        current_page = getPage(current_page->getNextPageId());
        current_index = 0;
    }

    return update_count;
}

uint32_t BPlusTree::UpdateAll(const std::unordered_map<std::string, Value> &new_values, CatalogManager &catalog_manager_)
{
    // 1. 找到最左边的叶子页面
    uint32_t update_count = 0;
    auto first_leaf_page_id = findFirstLeafPageId();
    uint32_t current_page_id = first_leaf_page_id;
    if (current_page_id == INVALID_PAGE_ID)
    {
        return update_count; // 树为空
    }
    auto current_page = getPage(current_page_id);
    int num_records = current_page->getNumRecords();
    // 2. 顺着叶子节点的双向链表遍历
    while (current_page)
    {
        num_records = current_page->getNumRecords();
        // 提取当前页面的所有行并更新
        for (int i = 0; i < num_records; ++i)
        {
            update_count += UpdateInternal(current_page.get(), i, new_values, catalog_manager_);
        }
        StorageSystem::write_page(current_page->getPageId(), current_page->serialize());

        // 移动到下一个页面
        uint32_t next_page_id = current_page->getNextPageId();
        // 检查是否遍历完成（到达链表末尾，或者在只有根节点的树中回到自身）
        if (next_page_id == INVALID_PAGE_ID || next_page_id == first_leaf_page_id)
        {
            break;
        }
        current_page = getPage(next_page_id);
    }

    return update_count;
}

uint32_t BPlusTree::UpdateInternal(DBPage *page, int index, const std::unordered_map<std::string, Value> &new_values, CatalogManager &catalog_manager_)
{
    uint32_t update_count = 0;
    Row row;
    if (!page->getLeafValue(index, row))
    {
        return update_count; // 获取行失败
    }
    TableSchema schema = catalog_manager_.getTableSchema(table_name_);
    std::vector<ColumnSchema> columns = schema.cols;
    for (const auto &[column, value] : new_values)
    {
        auto it = std::find_if(columns.begin(), columns.end(), [&](const ColumnSchema &col)
                               { return col.name == column; });
        if (it != columns.end())
        {
            int64_t index = std::distance(columns.begin(), it);
            row[index] = value;
        }
    }
    page->setLeafValue(index, row);
    // StorageSystem::write_page(page->getPageId(), page->serialize());
    update_count++;
    return update_count;
}